package cn.sciento.transfer.infra.rocketmq.handler;


import java.nio.charset.StandardCharsets;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;


import cn.sciento.core.convertor.ApplicationContextHelper;
import cn.sciento.core.exception.CommonException;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import cn.sciento.starter.lock.LockInfo;
import cn.sciento.starter.lock.service.LockService;
import cn.sciento.starter.lock.service.impl.ReentrantLockServiceImpl;
import cn.sciento.transfer.app.service.ConsumerEventService;
import cn.sciento.transfer.domain.entity.ConsDbConfig;
import cn.sciento.transfer.domain.entity.ConsTenantConfig;
import cn.sciento.transfer.infra.datasource.sql.FieldBuilder;
import cn.sciento.transfer.infra.datasource.sql.SqlUtils;
import cn.sciento.transfer.infra.datasource.utils.TransferDataSourceTemplate;
import cn.sciento.transfer.infra.rocketmq.vo.MqMessage;
import cn.sciento.transfer.infra.rocketmq.vo.MqMessageSqlMapping;
import cn.sciento.transfer.infra.rocketmq.vo.SqlInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;


public class MqMessageHandler implements MessageListenerConcurrently {
    private Logger LOGGER = LoggerFactory.getLogger(MqMessageHandler.class);

    private ConsDbConfig consDbConfig;

    private ConsTenantConfig consTenantConfig;

    private ConsumerEventService consumerEventService;

    private TransferDataSourceTemplate transferDataSourceTemplate;

    public MqMessageHandler(ConsumerEventService consumerEventService, TransferDataSourceTemplate databaseOperateService, ConsDbConfig consDbConfig, ConsTenantConfig consTenantConfig) {
        this.consumerEventService = consumerEventService;
        this.transferDataSourceTemplate = databaseOperateService;
        this.consDbConfig = consDbConfig;
        this.consTenantConfig = consTenantConfig;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        if (CollectionUtils.isEmpty(msgs)) {
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        if (this.LOGGER.isDebugEnabled()) {
            this.LOGGER.debug("消费线程 = {}，消费消息数量 = {}",
                    Thread.currentThread().getName() + Thread.currentThread().getId(),
                    msgs.size());
        }
        for (MessageExt messageExt : msgs) {
            if (this.LOGGER.isDebugEnabled()) {
                this.LOGGER.debug("来源topic = {},消费组 = {},\n消费消息 = {}", messageExt
                        .getTopic(), this.consDbConfig
                        .generateConsumerGroup(this.consTenantConfig), new String(messageExt
                        .getBody(), StandardCharsets.UTF_8));
            }
        }
        List<MqMessage> uniqueMqMessagesSorted = null;
        try {
            initCondition(this.consDbConfig);
            uniqueMqMessagesSorted = preHandleMsg(msgs);
        } catch (Exception e) {
            if (this.LOGGER.isDebugEnabled()) {
                this.LOGGER.debug(">>>>>>>>>>some bad error happen ,consumer later....<<<<<<<<<", e);
            }
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        if (CollectionUtils.isEmpty(uniqueMqMessagesSorted)) {
            if (this.LOGGER.isDebugEnabled()) {
                this.LOGGER.debug("============ no msg need to process ===========");
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        if (this.LOGGER.isDebugEnabled()) {
            this.LOGGER.debug("=============实际上处理的消息量 [{}]=================",
                    uniqueMqMessagesSorted.size());
        }
        try {
            List<MqMessage> finalUniqueMqMessagesSorted = uniqueMqMessagesSorted;
            this.transferDataSourceTemplate.doExecute(this.consDbConfig.getConsumerDs(), this.consDbConfig.getConsumerService(), jdbcTemplate -> {
                consumerMsg(finalUniqueMqMessagesSorted, true, jdbcTemplate);
                return null;
            });
        } catch (Exception e) {
            if (this.LOGGER.isDebugEnabled()) {
                this.LOGGER.debug("consumer error,configDbId[{}],configService[{}],configTable[{}],\n{}", this.consDbConfig
                                .getConsDbConfigId(), this.consDbConfig
                                .getConsumerService(), this.consDbConfig
                                .getConsumerDb(),
                        ExceptionUtils.getStackTrace(e));
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    public List<MqMessage> preHandleMsg(List<MessageExt> msgs) {
        try {
            List<MqMessage> mqMessageList = MqMessage.convertMessageExtToMqMessage(msgs);
            List<MqMessage> plainMsgList = new ArrayList<>();
            for (MqMessage msg : mqMessageList) {
                plainMsgList.addAll(msg.plainMsg());
            }
            return persistExceptRepeatMsgAndLoadAdvanceMsg(plainMsgList, this.consDbConfig);
        } catch (Exception e) {
            throw new CommonException(e);
        }
    }

    public void consumerMsg(List<MqMessage> mqMessageList, boolean errorReport, JdbcTemplate jdbcTemplate) {
        for (MqMessage mqMessage : mqMessageList) {
            try {
                this.consumerEventService.updateMessageStatus(this.consDbConfig, mqMessage, "R", "", jdbcTemplate);
                List<MqMessageSqlMapping> messageSqlMappingList = MqMessageSqlMapping.builder().builderSqlInfo(mqMessage);
                distributedSyncExecute(lockSuccess -> {
                    if (this.consumerEventService.checkMqMessageIsExpired(this.consDbConfig, mqMessage, jdbcTemplate)) {
                        consumerEventArch(mqMessage, jdbcTemplate);
                        return;
                    }
                    switch (mqMessage.getAction()) {
                        case "INSERT":
                            for (MqMessageSqlMapping mqMessageSqlMapping : messageSqlMappingList) {
                                try {
                                    executeSqlMapping(mqMessageSqlMapping, mqMessage.getTable(), jdbcTemplate);
                                } catch (SQLException e) {
                                    throw new CommonException(e);
                                }
                            }
                            break;
                        case "UPDATE":
                            for (MqMessageSqlMapping mqMessageSqlMapping : messageSqlMappingList) {
                                Map<String, Object> rowData = mqMessageSqlMapping.getRowData();
                                if (!dataExistsById(jdbcTemplate, this.consDbConfig.getConsumerFullTableName(), mqMessage.getKeyField(), rowData.get(mqMessage.getKeyField()))) {
                                    if (this.LOGGER.isDebugEnabled()) {
                                        this.LOGGER.debug("================UPDATE DATA NOT EXISTS:[id={}]=================", rowData.get(mqMessage.getKeyField()));
                                    }
                                    this.consumerEventService.updateMessageStatus(this.consDbConfig, mqMessage, "N", "", jdbcTemplate);
                                    return;
                                }
                                try {
                                    executeSqlMapping(mqMessageSqlMapping, mqMessage.getTable(), jdbcTemplate);
                                } catch (SQLException e) {
                                    throw new CommonException(e);
                                }
                            }
                            break;
                        case "DELETE":
                            for (MqMessageSqlMapping mqMessageSqlMapping : messageSqlMappingList) {
                                try {
                                    executeSqlMapping(mqMessageSqlMapping, mqMessage.getTable(), jdbcTemplate);
                                } catch (SQLException e) {
                                    throw new CommonException(e);
                                }
                            }
                            break;
                    }
                    consumerEventArch(mqMessage, jdbcTemplate);
                    List<MqMessage> preMsg = this.consumerEventService.loadNMsgListLessThanEventId(this.consDbConfig, mqMessage, jdbcTemplate);
                    for (MqMessage msg : preMsg) {
                        consumerEventArch(msg, jdbcTemplate);
                    }
                });
            } catch (Exception e) {
                try {
                    this.consumerEventService.updateMessageStatus(this.consDbConfig, mqMessage, "E",

                            ExceptionUtils.getMessage(e), jdbcTemplate);
                } catch (Exception e1) {
                    if (this.LOGGER.isDebugEnabled()) {
                        this.LOGGER.debug(e1.getMessage(), e);
                    }
                }
                if (errorReport) {
                    errorReport(mqMessage, e, jdbcTemplate);
                } else {
                    this.consumerEventService.updateMessageStatus(this.consDbConfig, mqMessage, "ER",

                            ExceptionUtils.getMessage(e), jdbcTemplate);
                }
                throw new CommonException(e);
            }
        }
    }

    private void errorReport(MqMessage mqMessage, Throwable e, JdbcTemplate jdbcTemplate) {
        try {
            this.consumerEventService.errorHandler(mqMessage, this.consDbConfig,

                    ExceptionUtils.getMessage(e));
        } catch (Exception e1) {
            this.consumerEventService.updateMessageStatus(this.consDbConfig, mqMessage, "ERE",

                    ExceptionUtils.getMessage(e1), jdbcTemplate);
            return;
        }
        this.consumerEventService.updateMessageStatus(this.consDbConfig, mqMessage, "ER",

                ExceptionUtils.getMessage(e), jdbcTemplate);
    }

    private void consumerEventArch(MqMessage msg, JdbcTemplate jdbcTemplate) {
        this.consumerEventService.updateMessageStatus(this.consDbConfig, msg, "S", "", jdbcTemplate);
        this.consumerEventService.moveMqMessageToArchTable(this.consDbConfig, msg, jdbcTemplate);
    }

    private List<MqMessage> persistExceptRepeatMsgAndLoadAdvanceMsg(List<MqMessage> mqMessageList, ConsDbConfig consDbConfig) {
        return this.transferDataSourceTemplate.doExecuteInTransaction(consDbConfig.getConsumerDs(), consDbConfig.getConsumerService(), (jdbcTemplate) -> {
            List<MqMessage> filterRepeatMsgList = mqMessageList.stream().filter((item) -> !this.consumerEventService.archMessageIfExists(consDbConfig, item, jdbcTemplate)).sorted(MqMessage.compareEventId(MqMessage.class)).peek((item) -> {
                item.setTenantId(this.consTenantConfig != null ? this.consTenantConfig.getTenantId() : null);
            }).collect(Collectors.toList());
            this.consumerEventService.saveMessageToDb(consDbConfig, filterRepeatMsgList, jdbcTemplate);
            MqMessage minEventIdMsgEntity = (mqMessageList.stream()
                    .sorted(MqMessage.compareEventId(MqMessage.class)).collect(Collectors.toList())).get(0);
            List<MqMessage> preConsumeMsgList = new ArrayList();
            this.distributedSyncExecute((locSuccess) -> {
                if (!locSuccess) {
                    throw new CommonException("acquire redis lock failed");
                } else {
                    preConsumeMsgList.addAll(this.consumerEventService.loadNMsgListLargeThanEventId(consDbConfig, minEventIdMsgEntity, jdbcTemplate));

                    for (MqMessage pMsg : preConsumeMsgList) {
                        this.consumerEventService.updateMessageStatus(consDbConfig, pMsg, "R", "", jdbcTemplate);
                    }

                }
            });
            filterRepeatMsgList.addAll(preConsumeMsgList);
            return filterRepeatMsgList;
        });
    }

    private int executeSqlMapping(MqMessageSqlMapping mqMessageSqlMapping, String produceTableName, JdbcTemplate jdbcTemplate) throws SQLException {
        SqlInfo<String, List<Object>> sqlInfo = (SqlInfo<String, List<Object>>) mqMessageSqlMapping.getSqlMapping();
        String sql = sqlInfo.getSql().replace(produceTableName, this.consDbConfig.getConsumerFullTableName());
        int result = jdbcTemplate.update(sql, sqlInfo.getData().toArray());
        if (this.consDbConfig.getMultiLangConsumerFlag()) {
            List<SqlInfo<?, ?>> tlSqlInfoList = mqMessageSqlMapping.getTlSqlInfoList();
            if (CollectionUtils.isNotEmpty(tlSqlInfoList)) {

                for (SqlInfo<?, ?> info : tlSqlInfoList) {
                    SqlInfo item = info;
                    String tlSql = ((String) item.getSql()).replace(produceTableName, this.consDbConfig.getConsumerFullTableTLName());
                    jdbcTemplate.update(tlSql, ((List) item.getData()).toArray());
                }
            }
        }

        return result;
    }

    public boolean dataExistsById(JdbcTemplate jdbcTemplate, String table, String idField, Object idValue) {
        String selectSql = String.format("SELECT COUNT(*) FROM %s WHERE %s", table, FieldBuilder.formatFieldToColumn(idField) + " = ?");
        List<Integer> count = jdbcTemplate.queryForList(selectSql, Integer.class, idValue);
        return (CollectionUtils.isNotEmpty(count) && count.get(0) > 0);
    }

    public void initCondition(ConsDbConfig consDbConfig) {
        this.transferDataSourceTemplate.doExecuteInConnection(consDbConfig.getConsumerDs(), consDbConfig
                .getConsumerService(), consDbConfig
                .getConsumerDb(), connection -> {
            consDbConfig.setConsumerEvent(SqlUtils.buildQualifiedTableNameInConnection(connection, consDbConfig.getConsumerDb(), "hdtt_consumer_event"));
            consDbConfig.setConsumerEventArch(SqlUtils.buildQualifiedTableNameInConnection(connection, consDbConfig.getConsumerDb(), "hdtt_consumer_event_arch"));
            consDbConfig.setConsumerFullTableName(SqlUtils.buildQualifiedTableNameInConnection(connection, consDbConfig.getConsumerDb(), consDbConfig.getConsumerTable()));
            consDbConfig.setConsumerFullTableTLName(SqlUtils.buildQualifiedTableNameInConnection(connection, consDbConfig.getConsumerDb(), consDbConfig.getConsumerTlTable()));
            consDbConfig.setMultiLangConsumerFlag(SqlUtils.checkTableExistsWithInOnceStatement(connection, consDbConfig.getConsumerTlTable()));
            return null;
        });
    }

    private void distributedSyncExecute(Consumer<Boolean> consumer) {
        String consumerGroup = this.consDbConfig.generateConsumerGroup(this.consTenantConfig);
        LockService lockService = ApplicationContextHelper.getContext().getBean(ReentrantLockServiceImpl.class);
        lockService.setLockInfo(new LockInfo(consumerGroup,
                Collections.emptyList(), 300L, 300L, TimeUnit.SECONDS));
        try {
            boolean lockSuccess = lockService.lock();
            consumer.accept(lockSuccess);
        } finally {
            lockService.releaseLock();
        }
    }
}
